﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using NetMQ;
using NetMQ.Sockets;

namespace Lawn.NetMessager
{
    public class MessageEventArgs : EventArgs
    {

        public byte[] Data { get; }
        public string Topic { get; }
        public MessageEventArgs(byte[] data, string topic)
        {
            this.Data = data;
            this.Topic = topic;
        }
    }
    public class MessageReciver:IDisposable
    {



        public string Url { get; }
        public int  Port { get; }

        public string[] Topics { get; }

        

        internal SubscriberSocket SubscriberSocket { get; }

        protected NetMQPoller Poller { get; }


        protected NetMQTimer HeatTimer { get; }

       


      



        public MessageReciver(string url,int port,string[] topics) : this(topics)
        {
            this.Url = url;
            this.Port = port;
         
        }

        public MessageReciver(string[] topics = null)
        {
            this.Topics = topics;

            SubscriberSocket = new SubscriberSocket();
            HeatTimer = new NetMQTimer(TimeSpan.FromMilliseconds(CONSTANT.HEAT_TIME));
            Poller = new NetMQPoller(); 
            Poller.Add(SubscriberSocket);
            Poller.Add(HeatTimer);

            this.HeatTimer.Elapsed += HeatTimer_Elapsed;
            this.SubscriberSocket.ReceiveReady += SubscriberSocket_ReceiveReady;


            Subscribe(true);
           

        }


        private void Subscribe(bool sub)
        {
            if (sub)
            {
                if (Topics != null)
                {
                    foreach (var topic in Topics)
                    {
                        this.SubscriberSocket.Subscribe(topic);
                    }
                }

                this.SubscriberSocket.Subscribe(CONSTANT.HEAT_BEAT);

            }
            else
            {
                if (Topics != null)
                {
                    foreach (var topic in Topics)
                    {
                        this.SubscriberSocket.Unsubscribe(topic);
                    }
                }

                this.SubscriberSocket.Unsubscribe(CONSTANT.HEAT_BEAT);
            }

        }




        private void HeatTimer_Elapsed(object sender, NetMQTimerEventArgs e)
        {


            if (HeatData.Count > 0)
            {
                HeatData.Clear();
                this.OnFireConnected();
            }
            else
            {
                OnFireConnectError();
            }
            

        }

        protected List<long> HeatData = new List<long>();

        private void MessageHandler_ConnectError(object sender, EventArgs e)
        {
            this.OnFireConnectError();
        }

        protected void FireMessageRecive(byte[] data, string topic)
        {
            if (messageRecive != null)
            {
                messageRecive(this, new MessageEventArgs(data, topic));
            }
        }
        protected event EventHandler<MessageEventArgs> messageRecive;

        public event EventHandler<MessageEventArgs> MessageRecive
        {
            add
            {
                messageRecive += value;
            }
            remove
            {
                messageRecive -= value;
            }
        }


        private event EventHandler connected;

        public event EventHandler Connected
        {
            add
            {
                this.connected += value;
            }
            remove
            {
                this.connected -= value;
            }
        }

        protected void OnFireConnected()
        {
            if (this.connected != null)
            {
                this.connected(this, EventArgs.Empty);
            }
        }

        private event EventHandler connectError;

        public event EventHandler ConnectError
        {
            add
            {
                this.connectError += value;
            }
            remove
            {
                this.connectError -= value;
            }
        }
       
        protected void OnFireConnectError()
        {
            if (this.connectError != null)
            {
                this.connectError(this, EventArgs.Empty);
            }
        }


         

        private void SubscriberSocket_ReceiveReady(object sender, NetMQSocketEventArgs e)
        {
            if (!this.Poller.IsRunning) return;
            if (!e.IsReadyToReceive) return;
            bool more;
            var topic = e.Socket.ReceiveFrameString(out more);
            byte[] arr_request = null; 
            if (more)
            { 
                var received = e.Socket.TryReceiveFrameBytes(TimeSpan.FromSeconds(1), out arr_request);
               
            }
            if (topic == CONSTANT.HEAT_BEAT)
            {
                HeatData.Add(1);
            }
            else
            { 
                FireMessageRecive(arr_request, topic); 
            }
           
          

        }

        public void Start ()
        {
            this.SubscriberSocket.Connect($"tcp://{this.Url}:{this.Port}");

            if (!this.Poller.IsRunning)
            {
                this.Poller.RunAsync();
            }
          
        }

        public Task Stop()
        {


          return  Task.Factory.StartNew(() =>
          {
              this.Poller.StopAsync();

              Subscribe(false);
                HeatTimer.Enable = true;
                this.Poller.Remove(HeatTimer);
               
                this.Poller.Remove(SubscriberSocket);

               //this.Poller.Dispose();

               // NetMQConfig.Cleanup();
            });
             

           



        }

        public void Dispose()
        {
            this.Stop();
        }
    }
}
